package online_mqtt

import (
	"errors"
	"fmt"
	"regexp"
	"server/global"
	"server/global/mqtt/fsc"
	"server/model"
	"server/service"
	"server/utils"
	"time"

	MQTT "github.com/eclipse/paho.mqtt.golang"
	"go.uber.org/zap"
)

// 设备在线状态处理
func OnlineDeal(sn string, connect string, client MQTT.Client) error {
	// 根据sn查询设备信息
	device := &model.Device{Sn: sn}
	exist, _ := service.DeviceExistBySn(*device)
	if !exist {
		return errors.New(fmt.Sprintf("[%v]设备不存在, 无法更新在线状态~", sn))
	}
	device, err := service.DeviceGetBySn(*device)
	if err != nil {
		return err
	}
	// 查询到设备存在于数据库中
	if connect == "connected" {
		// global.LOG.Debug(fmt.Sprintf("[%v]上线", sn))
		// 发送上线消息
		info := fmt.Sprintf(`{"aswcode":"%v","aswmsg":"%v","clientid":"%v"}`, fsc.CODE_SYSTEM_ONLINE, 1, device.Sn)
		client.Publish("fsc/online", 0, false, info)
		// 修改数据库中状态
		device.Online = 1
		err = service.DeviceEdit(*device)
		if err != nil {
			return err
		}
		// 添加记录
		dlog := &model.DeviceLog{DeviceID: device.ID, Value: "上线", Status: 0}
		err = service.DeviceLogAdd(*dlog)
		if err != nil {
			return err
		}
	}
	if connect == "disconnected" {
		// global.LOG.Debug(fmt.Sprintf("[%v]掉线", sn))
		// 发送掉线消息
		info := fmt.Sprintf(`{"aswcode":"%v","aswmsg":"%v","clientid":"%v"}`, fsc.CODE_SYSTEM_ONLINE, 0, device.Sn)
		client.Publish("fsc/online", 0, false, info)
		// 修改数据库中状态
		device.Online = 0
		err = service.DeviceEdit(*device)
		if err != nil {
			return err
		}
		// 添加记录
		dlog := &model.DeviceLog{DeviceID: device.ID, Value: "离线", Status: 0}
		err = service.DeviceLogAdd(*dlog)
		if err != nil {
			return err
		}
	}
	return nil
}

var messageHandler MQTT.MessageHandler = func(client MQTT.Client, msg MQTT.Message) {
	// fmt.Println("************************** online message ****************************")
	// fmt.Printf("topic: %s\n", msg.Topic())
	// fmt.Printf("message: %s\n", msg.Payload())
	// global.LOG.Info("***** ONLINE *****")
	// 匹配终端设备上线, sn编码为10位的编码
	reg_connect, err := regexp.Compile(`^\$SYS/brokers/(.*?)/clients/([a-zA-Z0-9]{0,100})/(connected|disconnected)`)
	if err != nil {
		global.LOG.Error(fmt.Sprintf("%v", err))
		return
	}
	if reg_connect.MatchString(msg.Topic()) == true {
		result := reg_connect.FindStringSubmatch(msg.Topic())
		sn := result[2]
		connect := result[3]
		global.LOG.Info("online", zap.Any("sn", sn), zap.Any("connect", connect))
		err = OnlineDeal(sn, connect, client)
		if err != nil {
			global.LOG.Error(fmt.Sprintf("%v", err))
		}
	}
}

type Client struct {
	MQTT.Client
	Host     string
	ClientID string
	UserName string
	Password string
}

type Message struct {
	ClientID string `json:"clientId"`
	Type     string `json:"type"`
	Data     string `json:"data,omitempty"`
	Time     int64  `json:"time"`
}

func (m *Client) NewClient() {
	ClientID := "server-online-" + utils.GenerateTimeMD5()[0:8]
	opts := MQTT.NewClientOptions()
	opts.AddBroker(global.CONFIG.Emqx.Host)
	opts.SetClientID(ClientID)
	opts.SetUsername(global.CONFIG.Emqx.Username)
	opts.SetPassword(global.CONFIG.Emqx.Password)
	opts.SetCleanSession(true)
	opts.SetAutoReconnect(true)
	opts.SetKeepAlive(120 * time.Second)
	opts.SetPingTimeout(10 * time.Second)
	opts.SetWriteTimeout(10 * time.Second)
	opts.SetOnConnectHandler(func(client MQTT.Client) {
		// 连接被建立后的回调函数
		global.LOG.Info(fmt.Sprintf(" Mqtt is connected! clientId: %v", ClientID))
		// 重连后重新订阅
		m.Subscribe()
	})
	opts.SetConnectionLostHandler(func(client MQTT.Client, err error) {
		// 连接被关闭后的回调函数
		global.LOG.Info(fmt.Sprintf("Mqtt is disconnected! clientId: %v reason: %v", ClientID, err.Error()))
	})

	m.Client = MQTT.NewClient(opts)
}

func (m *Client) Connect() {
	global.LOG.Info("connect")

	token := m.Client.Connect()
	if token.Wait() && token.Error() != nil {
		global.LOG.Error(fmt.Sprintf("%v", token.Error()))
	}
}

func (m *Client) DisConnect() {
	fmt.Println("disconnect")
	m.Client.Disconnect(250)
}

func (m *Client) Subscribe() {
	global.LOG.Info("subscribe")
	topics := make(map[string]byte)
	// topics["$SYS/brokers/+/clients/#"] = 0
	for _, v := range global.CONFIG.Emqx.OnlineTopics {
		topics[v] = 0
	}
	global.LOG.Info(fmt.Sprintf("%v", topics))
	// token := m.Client.Subscribe("dqt201/+/tx", 0, messageHandler)
	token := m.Client.SubscribeMultiple(topics, messageHandler)
	if token.Wait() && token.Error() != nil {
		global.LOG.Error(fmt.Sprintf("%v", token.Error()))
	}
}

func (m *Client) Pub() {
	global.LOG.Info("publish")
	s := time.Now().String()
	token := m.Client.Publish("date", 0, false, s)
	if token.Wait() && token.Error() != nil {
		global.LOG.Error(fmt.Sprintf("%v", token.Error()))
	}
}

func (m *Client) Run() {
	m.NewClient()
	m.Connect()
}
